package com.zst.amqp;

import com.zst.amqp.entity.MessageDTO;
import com.zst.amqp.entity.PublisherCorrelationData;
import com.zst.amqp.event.PublisherConfirmCallback;
import com.zst.amqp.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.MessagePropertiesBuilder;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

/**
 * 用于发送消息到消息队列的类
 *
 * 发送消息给RabbitMQ时使用的RabbitTemplate会等到调用send方法的时候
 */
public class MessageSender {
    private static final Logger logger = LoggerFactory.getLogger(MessageSender.class);
    private RabbitTemplate rabbitTemplate;
    private String exchangeName;

    public MessageSender(ConnectionFactory connectionFactory,
                         PublisherConfirmCallbackContext confirmCallbackContext,
                         String exchangeName) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        rabbitTemplate.setConfirmCallback(confirmCallback(confirmCallbackContext));
        rabbitTemplate.setReturnCallback(returnCallback());
        this.rabbitTemplate = rabbitTemplate;
        this.exchangeName = exchangeName;
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param type 消息的类型
     */
    public void send(String content, String type) {
        send(content, type, "", "");
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param exchangeName 交换器名称
     * @param type 消息的类型
     */
    public void send(String content, String type, String exchangeName) {
        send(content, type, exchangeName, "", "");
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param type 消息类型
     * @param correlationId 消息ID，用于异步回调确认时查找Message数据
     * @param confirmCallbackKey 消息确认回调函数的key
     */
    public void send(String content, String type, String correlationId, String confirmCallbackKey) {
        send(content, type, this.exchangeName, correlationId, confirmCallbackKey);
    }

    /**
     * 发送消息
     * @param content 消息内容
     * @param type 消息类型
     * @param exchangeName 交换器名称
     * @param correlationId 消息ID，用于异步回调确认时查找Message数据
     * @param confirmCallbackKey 消息确认回调函数的key
     */
    public void send(String content, String type, String exchangeName, String correlationId, String confirmCallbackKey) {
        // 把消息内容包装成MessageDTO
        MessageDTO dto = MessageDTO.create(content, type);

        // 设置消息属性为持久化消息
        MessageProperties prop = MessagePropertiesBuilder.newInstance()
                .setDeliveryMode(MessageDeliveryMode.PERSISTENT)
                .build();

        // 创建消息实例和发送确认数据
        Message message = MessageBuilder.withBody(JsonUtils.objectToJsonString(dto).getBytes()).andProperties(prop).build();
        PublisherCorrelationData publisherCorrelationData = PublisherCorrelationData.build(correlationId, confirmCallbackKey);

        // 通过rabbitTemplate，把消息发送出去
        this.rabbitTemplate.send(
                this.exchangeName,
                type,
                message,
                publisherCorrelationData);
    }

    /**
     * 发送消息
     * @param message
     * @param type
     * @param exchangeName
     */
    public void send(Message message, String type, String exchangeName) {
        this.rabbitTemplate.send(exchangeName, type, message);
    }

    /**
     * 提供给rabbitTemplate的ConfirmCallback
     * 如果confirmCallback上下文里有消息对应的回调对象的话，就调用，没有就打印一条日志
     *
     * @return
     */
    private RabbitTemplate.ConfirmCallback confirmCallback(PublisherConfirmCallbackContext confirmCallbackContext) {
        return (correlationData, ack, cause) -> {
            // 如果correlationData为空，说明消息发送时没有添加回调信息，忽略
            if (correlationData == null) {
                return;
            }

            PublisherCorrelationData publisherCorrelationData = (PublisherCorrelationData) correlationData;
            PublisherConfirmCallback callback = confirmCallbackContext.getCallbackFunction(publisherCorrelationData.getKey());
            if (callback != null) {
                //找到对应的回调函数，调用
                callback.confirmCallback(publisherCorrelationData.getId(), ack);
            } else {
                //没有回调函数，则使用默认的处理逻辑
                if (ack) {
                    logger.debug("收到消息队列收到消息的确认，消息确认id为" + correlationData.getId());
                } else {
                    logger.error("收到消息队列对发出的消息拒绝接受的确认，原因为:" + cause);
                }
            }
        };
    }

    /**
     * 返回一个默认的ReturnCallback对象
     * @return
     */
    private RabbitTemplate.ReturnCallback returnCallback() {
        return (message, replyCode, replyText, exchange, routingKey) -> {

        };
    }
}
